package com.nageoffer.shortlink.project.mq.consumer;

import com.nageoffer.shortlink.project.common.convention.exception.ServiceException;
import com.nageoffer.shortlink.project.dto.biz.ShortLinkStatsRecordDTO;
import com.nageoffer.shortlink.project.mq.idempotent.MessageQueueIdempotentHandler;
import com.nageoffer.shortlink.project.service.ShortLinkService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import java.util.concurrent.Executors;
import java.util.concurrent.locks.LockSupport;

import static com.nageoffer.shortlink.project.common.constant.RedisKeyConstant.DELAY_QUEUE_STATS_KEY;


@Slf4j
@Component
@RequiredArgsConstructor
public class DelayShortLinkStatsConsumer implements InitializingBean {

    // Redisson客户端，用于操作Redis分布式数据结构
    private final RedissonClient redissonClient;
    // 短链接服务，用于处理统计逻辑
    private final ShortLinkService shortLinkService;
    // 消息队列幂等处理器，防止重复消费
    private final MessageQueueIdempotentHandler messageQueueIdempotentHandler;

    // 消息消费主方法
    public void onMessage() {
        // 创建单线程执行器（守护线程）
        Executors.newSingleThreadExecutor(
                        runnable -> {
                            Thread thread = new Thread(runnable);
                            thread.setName("delay_short-link_stats_consumer");  // 设置线程名称
                            thread.setDaemon(Boolean.TRUE);  // 设置为守护线程
                            return thread;
                        })
                .execute(() -> {
                    // 获取Redis阻塞队列实例
                    RBlockingDeque<ShortLinkStatsRecordDTO> blockingDeque = redissonClient.getBlockingDeque(DELAY_QUEUE_STATS_KEY);
                    // 获取与阻塞队列关联的延迟队列实例
                    RDelayedQueue<ShortLinkStatsRecordDTO> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
                    // 无限循环处理消息
                    for (; ; ) {
                        try {
                            // 从延迟队列中获取消息（非阻塞）
                            ShortLinkStatsRecordDTO statsRecord = delayedQueue.poll();
                            if (statsRecord != null) {
                                // 检查消息是否已被处理（幂等控制）
                                if (!messageQueueIdempotentHandler.isMessageProcessed(statsRecord.getKeys())) {
                                    // 若未被处理，检查是否已完成流程
                                    if (messageQueueIdempotentHandler.isAccomplish(statsRecord.getKeys())) {
                                        return;  // 已完成则直接返回
                                    }
                                    // 未完成流程则抛出异常，触发消息队列重试机制
                                    throw new ServiceException("消息未完成流程，需要消息队列重试");
                                }
                                try {
                                    // 执行短链接统计业务逻辑
                                    shortLinkService.shortLinkStats(null, null, statsRecord);
                                } catch (Throwable ex) {
                                    // 处理异常时清除幂等标记，允许消息重试
                                    messageQueueIdempotentHandler.delMessageProcessed(statsRecord.getKeys());
                                    // 记录错误日志
                                    log.error("延迟记录短链接监控消费异常", ex);
                                }
                                // 标记消息处理完成
                                messageQueueIdempotentHandler.setAccomplish(statsRecord.getKeys());
                                continue;
                            }
                            // 队列为空时暂停500毫秒（避免CPU空转）
                            LockSupport.parkUntil(500);
                        } catch (Throwable ignored) {
                            // 捕获所有异常防止线程终止
                        }
                    }
                });
    }

    // Spring Bean初始化完成后自动调用
    @Override
    public void afterPropertiesSet() throws Exception {
        // 启动消息监听
        onMessage();
    }
}